package cn.doitedu.hbase.filter;

import cn.doitedu.hbase.baseapi.QueryDemos;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.Iterator;

/**
 * @date: 2019/7/11
 * @site: www.doitedu.cn
 * @author: hunter.d 涛哥
 * @qq: 657270652
 * @description: hbase的各类过滤器使用示例代码
 */
public class FilterDemos {


    Connection conn = null;
    Table t1 = null;

    @Before
    public void init() throws Exception {

        //Configuration conf = new Configuration();  它会自动加载classpath中的core-site.xml core-default.xml hdfs-site.xml hdfs-default.xml yarn-default.xml yarn-site.xml mapred-default.xml mapred-site.xml
        Configuration conf = HBaseConfiguration.create(); // 不光会加载hadoop的配置，还会加载  hbase-site.xml

        // 客户端连接不需要指定具体的master或者regionserver地址，只要指定zookeeper地址就行
        conf.set("hbase.zookeeper.quorum", "spark01:2181,spark02:2181,spark03:2181");

        conn = ConnectionFactory.createConnection(conf);

        t1 = conn.getTable(TableName.valueOf("t1"));

    }


    /**
     * 可以自定义规则的行键过滤器
     */
    @Test
    public void rowFilter() throws IOException {

        Scan scan = new Scan();
        scan.withStartRow("p001".getBytes());
        scan.withStopRow("p003".getBytes());


        // 指定一个比较运算符
        CompareOperator op_less = CompareOperator.LESS;
        CompareOperator op_equal = CompareOperator.EQUAL;

        // 指定一个比较参考对象
        ByteArrayComparable prefixComparatormp = new BinaryPrefixComparator("p0014".getBytes());
        BinaryComparator binaryCmp = new BinaryComparator("p0014".getBytes());
        SubstringComparator subStrCmp = new SubstringComparator("1399");


        // 构造一个行键过滤器，传入比较运算符，和比较的参考对象
        Filter f1 = new RowFilter(op_less, prefixComparatormp);  // 条件：  数据的行键的前缀  <  "p00202"
        Filter f2 = new RowFilter(op_equal, subStrCmp);  // 条件： 行键中有一个="1399"字串的数据

        FilterList filterList = new FilterList();
        filterList.addFilter(f1);
        filterList.addFilter(f2);


        // 设置过滤器 ： 只需要行键以p00202开头的数据
        scan.setFilter(filterList);


        ResultScanner scanner = t1.getScanner(scan);
        Iterator<Result> iter = scanner.iterator();
        while (iter.hasNext()) {
            Result r = iter.next();
            while (r.advance()) {
                Cell cell = r.current();
                QueryDemos.printCellUtil(cell);
            }
        }

    }


    /**
     *  封装好的行键前缀过滤器
     * @throws IOException
     */
    @Test
    public void testPrefixFilter() throws IOException {
        Scan scan = new Scan();
        scan.withStartRow("p001".getBytes());
        scan.withStopRow("p003".getBytes());

        Filter f = new PrefixFilter("p0013421".getBytes());

        scan.setFilter(f);

        ResultScanner scanner = t1.getScanner(scan);
        printResult(scanner);

    }


    /**
     * 自由定制的列族过滤器
     * @throws IOException
     */
    @Test
    public void testFamilyFilter() throws IOException {

        Scan scan = new Scan();
        scan.withStartRow("p001".getBytes());
        scan.withStopRow("p003".getBytes());

        FamilyFilter familyFilter = new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator("base_info".getBytes()));
        FamilyFilter familyFilter2 = new FamilyFilter(CompareOperator.EQUAL, new BinaryPrefixComparator("base".getBytes()));
        FamilyFilter familyFilter3 = new FamilyFilter(CompareOperator.NO_OP, new SubstringComparator("ase"));
        scan.setFilter(familyFilter);

        ResultScanner scanner = t1.getScanner(scan);
        printResult(scanner);
    }


    /**
     *  封装好的列名前缀过滤器
     */
    @Test
    public void testColumnPrefixFilter() throws IOException {
        Scan scan = new Scan();
        scan.withStartRow("p001".getBytes());
        scan.withStopRow("p003".getBytes());


        ColumnPrefixFilter f = new ColumnPrefixFilter("na".getBytes());
        scan.setFilter(f);

        ResultScanner scanner = t1.getScanner(scan);
        printResult(scanner);
    }

    /**
     *  可自由定制的列值过滤器
     */
    @Test
    public void testValueFilter() throws IOException {

        Scan scan = new Scan();
        scan.withStartRow("p001".getBytes());
        scan.withStopRow("p003".getBytes());



        Filter f = new ValueFilter(CompareOperator.GREATER_OR_EQUAL, new BinaryComparator(Bytes.toBytes(9000)));
        scan.setFilter(f);

        ResultScanner scanner = t1.getScanner(scan);
        printResult(scanner);

    }





    private void printResult(ResultScanner scanner) {
        Iterator<Result> iter = scanner.iterator();
        while (iter.hasNext()) {
            Result r = iter.next();
            System.out.println();
            while (r.advance()) {
                Cell cell = r.current();
                QueryDemos.printCellUtil(cell);
            }
        }
    }



    @After
    public void cleanup() throws IOException {

        t1.close();
        conn.close();
    }


}
